package com.kitten.dataAlign.consumer;

import com.kitten.dataAlign.constants.MQConstants;
import com.kitten.dataAlign.constants.RedisKeyConstants;
import com.kitten.dataAlign.constants.TableConstants;
import com.kitten.dataAlign.domain.mapper.InsertRecordMapper;
import com.kitten.dataAlign.model.dto.LikeUnLikeNoteMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.Objects;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_data_align_" + MQConstants.TOPIC_COUNT_NOTE_LIKE,
        topic = MQConstants.TOPIC_COUNT_NOTE_LIKE
)
public class TodayNoteLikeIncrementData2DBConsumer implements RocketMQListener<String> {

    @Resource
    private TransactionTemplate transactionTemplate;


    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private InsertRecordMapper insertRecordMapper;

    @Value("${table.shards}")
    private int tableShards;

    @Override
    public void onMessage(String body) {
        log.info("点赞/取消点赞 ===> data align consumer receive message: {}", body);
        // 1. 布隆过滤器判断该日增量数据是否已经记录
        LikeUnLikeNoteMqDTO dto = JsonUtils.parseObject(body, LikeUnLikeNoteMqDTO.class);
        if (Objects.isNull(dto)) {
            return;
        }
        Long noteId = dto.getNoteId();
        Long noteCreatorId = dto.getNoteCreatorId();

        // 今日日期
        String date = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        // 布隆过滤器key
        String key = RedisKeyConstants.buildBloomUserNoteLikeListKey(date); // bloom:dataAlign:note:likes:20250101

        /**
         * 脚本传入 Key 和 noteId
         * Redis 校验 key 是否存在, 不存在就 新增 bloom:dataAlign:note:likes:date RedisKey, 并传入空值
         * 之后校验该日期的 bloom 中是否含有该 noteId 的记录
         */
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/noteLikeUnLike/bloom_today_note_like_check.lua")));
        script.setResultType(Long.class);
        Long result = redisTemplate.execute(script, Collections.singletonList(key), noteId);

        RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);

        // ---------------------- 笔记点赞数记录 ----------------------
        if (Objects.equals(result, 0L)) {   // bloom 中含有key, 但是不包含记录
            // 将日增量变更数据，分别写入两张表
            // - t_data_align_note_like_count_temp_日期_分片序号
            // - t_data_align_user_like_count_temp_日期_分片序号

            // 根据分片总数，取模，分别获取对应的分片序号
            long noteIdHashKey = noteId % tableShards;
            try {
                insertRecordMapper.insert2DataAlignNoteLikeCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);

            } catch (Exception e) {
                log.error("insert data align note like count temp table error", e);

            }
            // 入库后, 添加布隆过滤器中
            redisTemplate.execute(bloomAddScript, Collections.singletonList(key), noteId);

        }

        // ---------------------- 用户被点赞数记录 ----------------------
        // 当用户1 点赞 用户2的两篇不同笔记时, 需要记录两次用户被点赞记录, 由于数据库表中 user_id字段含有unique索引, 所以需要先判断是否已经记录过, 再插入
        String userLikeCountBloomKey = RedisKeyConstants.buildBloomUserLikeListKey(date);
        result = redisTemplate.execute(script, Collections.singletonList(userLikeCountBloomKey), noteCreatorId);
        if (Objects.equals(result, 0L)) {
            long noteCreatorIdHashKey = noteCreatorId % tableShards;

            try {
                insertRecordMapper.insert2DataAlignUserLikeCountTempTable(TableConstants.buildTableNameSuffix(date, noteCreatorIdHashKey), noteCreatorId);
            } catch (Exception e) {
                log.error("insert data align user like count temp table error", e);
            }
            // 入库后, 添加布隆过滤器中
            redisTemplate.execute(bloomAddScript, Collections.singletonList(userLikeCountBloomKey), noteCreatorId);

        }

    }
}
